/*******************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.        */
/*                                                                 */
/*  NAME      = amqp_core.c                                        */
/*  NOTE      =                                                    */
/*  DATE      = 2015/09/09 by geyh                                 */
/*******************************************************************/

#include <stdio.h>
#include <sys/time.h>

#include "frmem.h"
#include "frcomm.h"
#include "frhash2.h"
#include "frreport_comm.h"
#include "frct_comm.h"
#include "frbyte_buffer.h"
#include "amqp_core.h"
#include "amqp_parse.h"
#include "amqp_content.h"

#include "config.h"
#include "amqp_detect.h"

extern void  *g_p_apm_context;


/*******************************************************************************
*  FUNC     :  创建私有数据
*  ARGS     :  void
            :
*  RTN      :  p_st_private         成功
*           :  NULL                 失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC st_amqp_private_data* __amqp_private_create(void)
{
    st_amqp_private_data *p_st_private = NULL;

    p_st_private = (st_amqp_private_data*)frmalloc(sizeof(st_amqp_private_data));
    if(p_st_private == NULL)
        return NULL;

    memset(p_st_private, '\x00', sizeof(st_amqp_private_data));

    return p_st_private;

}

/*******************************************************************************
*  FUNC     :  创建session数据   存放st_amqp_private_data
*  ARGS     :  void
            :
*  RTN      :  p_st_session_private         成功
*           :  NULL                 失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC L7session_amqp_private* __amqp_session_private_create(void)
{
    L7session_amqp_private *p_st_session_private = NULL;

    p_st_session_private = (L7session_amqp_private*)frmalloc(sizeof(L7session_amqp_private));
    if(p_st_session_private == NULL)
        return NULL;

    memset(p_st_session_private, '\x00', sizeof(L7session_amqp_private));

    p_st_session_private->p_st_data = bytebuff_creat();
    if(p_st_session_private->p_st_data == NULL)
    {
        frfree(p_st_session_private);
        return NULL;
    }
    p_st_session_private->i_version = 0xff;

    frlist_header_init(&p_st_session_private->st_session_private);

    return p_st_session_private;

}


/*******************************************************************************
*  FUNC     :  session数据上报
*  ARGS     :  p_st_body_data                   (数据链表)
            :  p_st_session_private             (私有数据)
*  RTN      :  FR_SUCC         成功
*           :  FR_FAIL         失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC int __amqp_session_report(Frreport_body_data *p_st_body_data, L7session_amqp_private *p_st_session_private)
{
    int i_ret = 0;

    if(p_st_body_data == NULL || p_st_session_private == NULL)
        return FR_FAIL;

    if(p_st_session_private->st_login_timestamp.tv_sec)
    {
        if(p_st_session_private->p_user_name)
        {
            if(frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_SESSION_USERNAME, p_st_session_private->p_user_name) == FR_FAIL)
                return FR_FAIL;
        }
        else
        {
            if(frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_SESSION_USERNAME, "\0") == FR_FAIL)
                return FR_FAIL;
        }

        if(frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_SESSION_ID, p_st_session_private->arr_session_id) == FR_FAIL)
            return FR_FAIL;

        if(frreport_body_l7_add_timeval(p_st_body_data, APM_REPORT_KEY_SESSION_LOGIN_TIMESTAMP, &p_st_session_private->st_login_timestamp) == FR_FAIL)
            return FR_FAIL;
    }

    if(p_st_session_private->p_server)
    {
        i_ret = frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_SESSION_SERVER, p_st_session_private->p_server);
        if(i_ret == FR_FAIL)
            return FR_FAIL;
    }

    return FR_SUCC;

}

/*******************************************************************************
*  FUNC     :  title数据上报
*  ARGS     :  p_st_body_data                   (数据链表)
            :  p_st_session_private             (私有数据)
*  RTN      :  FR_SUCC         成功
*           :  FR_FAIL         失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC int __amqp_title_report(Frreport_body_data *p_st_body_data, st_amqp_private_data *p_st_private)
{
    int i_ret = 0;
    char arr_tmp[128];

    if(p_st_body_data == NULL || p_st_private == NULL)
        return FR_FAIL;

    if(p_st_private->p_content_type)
    {
        i_ret = frreport_body_l7_add_str(p_st_body_data, "content_type", p_st_private->p_content_type);
        if(i_ret == FR_FAIL)
            return FR_FAIL;
    }

    if(p_st_private->p_content_encoding)
    {
        i_ret = frreport_body_l7_add_str(p_st_body_data, "content_encoding", p_st_private->p_content_encoding);
        if(i_ret == FR_FAIL)
            return FR_FAIL;
    }

    if(p_st_private->p_title)
    {
        if(p_st_private->i_use_request)
        {
            if(p_st_private->is_topic)
                i_ret = frreport_body_l7_add_str(p_st_body_data, "request.mq.topic", p_st_private->p_title);
            else
                i_ret = frreport_body_l7_add_str(p_st_body_data, "request.mq.queue", p_st_private->p_title);
            if(i_ret == FR_FAIL)
                return FR_FAIL;

            snprintf(arr_tmp, sizeof(arr_tmp), "PUT %s", p_st_private->p_title);
        }
        else
        {
            if(p_st_private->is_topic)
                i_ret = frreport_body_l7_add_str(p_st_body_data, "notify.mq.topic", p_st_private->p_title);
            else
                i_ret = frreport_body_l7_add_str(p_st_body_data, "notify.mq.queue", p_st_private->p_title);
            if(i_ret == FR_FAIL)
                return FR_FAIL;

            snprintf(arr_tmp, sizeof(arr_tmp), "NOTIFY %s", p_st_private->p_title);
        }

        i_ret = frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, arr_tmp);
        if(i_ret == FR_FAIL)
            return FR_FAIL;
    }

    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  st_amqp_private_data数据销毁
*  ARGS     :  p_st_private             (私有数据)
*  RTN      :  FR_SUCC         成功
*           :  FR_FAIL         失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC int __amqp_session_private_destroy(st_amqp_private_data *p_st_private)
{
    if(p_st_private)
    {
        if(p_st_private->p_st_body_data_req)
            frreport_body_free(&p_st_private->p_st_body_data_req);

        if(p_st_private->p_st_body_data_resp)
            frreport_body_free(&p_st_private->p_st_body_data_resp);

        if(p_st_private->p_lst_comm_tmp)
            frreport_body_l7list_free(&p_st_private->p_lst_comm_tmp);

        if(p_st_private->p_lst_req_tmp)
            frreport_body_l7list_free(&p_st_private->p_lst_req_tmp);

        if(p_st_private->p_lst_mq_tmp)
            frreport_body_l7list_free(&p_st_private->p_lst_mq_tmp);

        if(p_st_private->p_title)
            frfree(p_st_private->p_title);

        if(p_st_private->p_content_type)
            frfree(p_st_private->p_content_type);

        if(p_st_private->p_content_encoding)
            frfree(p_st_private->p_content_encoding);

        if(p_st_private->p_consumer_tag)
            frfree (p_st_private->p_consumer_tag);

        if(p_st_private->p_queue)
            frfree (p_st_private->p_queue);

        if(p_st_private->p_exchange_type)
            frfree (p_st_private->p_exchange_type);

        if(p_st_private->p_exchange)
            frfree (p_st_private->p_exchange);

        if(p_st_private->p_routing_key)
            frfree (p_st_private->p_routing_key);

        frfree(p_st_private);

    }

    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  st_amqp_private_data获取
*  ARGS     :  p_lst_head             (私有数据存放链表)
*           :  p_i_id_first
*           :  p_i_id_last
*  RTN      :  p_st_private         成功
*           :  NULL                 失败
*           :
*  NOTE     :
*******************************************************************************/
FR_STATIC st_amqp_private_data* __session_data_get(Frlist_header *p_lst_head, int *p_i_id_first, int *p_i_id_last)
{
    st_amqp_private_data *p_st_private = NULL;
    Frlist_node *p_st_node = NULL;
    int i_id_first = 0, i_id_last = 0;;

    if(p_lst_head == NULL || p_i_id_first == NULL || p_i_id_last == NULL)
        return NULL;

    i_id_first = *p_i_id_first;
    i_id_last = *p_i_id_last;

    //查询 id
    p_st_node = FRLIST_FIRST_NODE(p_lst_head);
    while(p_st_node && p_st_node->data)
    {
        p_st_private = (st_amqp_private_data*)FRLIST_GET_NODE_DATA(p_st_node);
        if(p_st_private->i_id_last >= i_id_first && i_id_first >= p_st_private->i_id_first)
        {
            //私有数据获取
            //如果该私有数据对应的response都已经解析完
            if(i_id_last >= p_st_private->i_id_last)
            {
                frlist_node_apart(p_lst_head, p_st_node);
                p_st_private->i_refence = 0;
                *p_i_id_first = p_st_private->i_id_last;
                frfree(p_st_node);
                return p_st_private;
            }
            else
            {
                *p_i_id_first = 0;
                *p_i_id_last = 0;
                return p_st_private;
            }
        }

    p_st_node = FRLIST_NEXT_NODE(p_st_node);

    }

    return NULL;
}

/*******************************************************************************
*  FUNC     :  判断是否是amqp
*  ARGS     :  p_st_conn                    输入数据信息
*           :  p_st_packet                 输入数据
*  RTN      :  APM_L7_IS_NO                 不是amqp数据
*           :  APM_L7_IS_YES                是amqp数据
*  NOTE     :
*******************************************************************************/
int amqp_detect(Conntrack *p_st_conn, Apm_packet *p_st_packet)
{
    int i_ret = 0;

    if(p_st_conn == NULL || p_st_packet == NULL)
        return APM_L7_IS_NO;

    i_ret = amqp_detect_0_9_0(p_st_conn, p_st_packet);
    if(i_ret == APM_L7_IS_YES)
        return APM_L7_IS_YES;

    i_ret = amqp_detect_1_0_0(p_st_conn, p_st_packet);
    if(i_ret == APM_L7_IS_YES)
        return APM_L7_IS_YES;

    return APM_L7_IS_NO;
}


/*******************************************************************************
*  FUNC     :  amqp req 的packet解析
*  ARGS     :  p_st_conn                (入力链接信息)
            :  p_st_packet             (入力packet信息)
            :  p_st_hash_report         (保存Frreport_data)
*  RTN      :  APM_L7_ANALYZE_SUCC         成功
*           :  APM_L7_ANALYZE_TODO         数据未结束
*           :  APM_L7_ANALYZE_FAIL         失败
*  NOTE     :
*******************************************************************************/
FR_STATIC int __req_data_analysis(Conntrack *p_st_conn
        , Apm_packet *p_st_packet
        , Frhash_header *p_st_hash_report
        , L7_session *p_st_session)
{
    int i_ret = 0;
    st_amqp_private_data *p_st_private = NULL;
    L7session_amqp_private *p_st_session_private = NULL;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data *p_st_report_data = NULL;
    unsigned long ul_tmp = 0;
    Frlist_node *p_st_node = NULL;
    Frlist_header *p_st_header = NULL;
    int i_falg_analysis = 1;

    if(p_st_conn == NULL || p_st_packet == NULL
       || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    p_st_session_private = (L7session_amqp_private*)p_st_session->p_session_private;

ANALYSIS:

    if(p_st_session_private->p_st_private == NULL)
    {
        p_st_session_private->p_st_private = __amqp_private_create();
        if(p_st_session_private->p_st_private == NULL)
            return APM_L7_ANALYZE_FAIL;
    }

    p_st_private = p_st_session_private->p_st_private;
    p_st_l4_trans = &(p_st_private->st_l4_trans);
    p_st_private->i_use_request = FR_YES;

    //第一个包  创建private 数据
    if(p_st_l4_trans->st_request_time_first_byte.tv_sec == 0
       && p_st_l4_trans->st_request_time_first_byte.tv_usec == 0)
    {
        p_st_private->p_st_body_data_req = frreport_body_new_l7(p_st_conn);
        if(p_st_private->p_st_body_data_req == NULL)
            return APM_L7_ANALYZE_FAIL;

        //st_request_time_first_byte
        p_st_l4_trans->st_request_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
        p_st_l4_trans->st_request_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;

        //seq ++
        p_st_private->ul_seq = frct_seq_increase(p_st_conn);
    }
    //st_request_time
    p_st_l4_trans->st_request_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
    p_st_l4_trans->st_request_time.tv_usec = p_st_packet->st_timestamp.tv_usec;

    if(i_falg_analysis)
    {
        i_ret = bytebuff_add(p_st_session_private->p_st_data, p_st_packet->p_payload, p_st_packet->ui_payload_len);
        if(i_ret == FR_FAIL)
            return APM_L7_ANALYZE_FAIL;
    }

    p_st_l4_trans->l_request_bytes = (long)p_st_session_private->p_st_data->ui_remaining;

    //如果skip长度有剩余  返回TODO
    if((p_st_session_private->p_st_data)->ui_skip_len)
        return APM_L7_ANALYZE_TODO;

    i_ret = amqp_parse(p_st_session_private->p_st_data, p_st_private->p_st_body_data_req, p_st_private, p_st_session_private);
    if(i_ret == APM_L7_ANALYZE_FAIL)
        return APM_L7_ANALYZE_FAIL;

    //session 数据处理
    if(p_st_session_private->i_flag_data_type == FRLOGIN_START)
    {
        p_st_session_private->st_login_timestamp.tv_sec = p_st_packet->st_timestamp.tv_sec;
        p_st_session_private->st_login_timestamp.tv_usec = p_st_packet->st_timestamp.tv_usec;

        fruuid(p_st_session_private->arr_session_id);
        p_st_session_private->i_flag_data_type = 0;
    }


    //数据上报
    if(i_ret == APM_L7_ANALYZE_SUCC && p_st_private->i_flag_report)
    {
        /* 请求的字节数 */
        p_st_l4_trans->l_request_bytes -= (long)p_st_session_private->p_st_data->ui_remaining;

        i_ret = frreport_body_l7_add_connInfo(p_st_private->p_st_body_data_req, p_st_conn);
        if(i_ret == FR_FAIL)
            return APM_L7_ANALYZE_FAIL;

        p_st_private->i_refence++;

        //session
        if(__amqp_session_report(p_st_private->p_st_body_data_req, p_st_session_private) == FR_FAIL)
            goto ERROR;

        if(__amqp_title_report(p_st_private->p_st_body_data_req, p_st_private) == FR_FAIL)
            goto ERROR;

        if(p_st_session_private->i_version == 10)
        {
            if(frreport_body_l7_add_int(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_ISRESPREQUIRED, 1) == FR_FAIL)
                goto ERROR;
        }
        else
        {
            if(frreport_body_l7_add_int(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_ISRESPREQUIRED, 0) == FR_FAIL)
                goto ERROR;
        }


        //信息上报
        if(frreport_body_l7_add_ulong(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_ID, p_st_private->ul_seq) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_START_TIMESTAMP, &p_st_l4_trans->st_request_time_first_byte) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_END_TIMESTAMP, &p_st_l4_trans->st_request_time) == FR_FAIL)
            goto ERROR;

        ul_tmp = (unsigned long)frtime_diff(&p_st_l4_trans->st_request_time_first_byte, &p_st_l4_trans->st_request_time);
        if(frreport_body_l7_add_ulong(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_SEND_TIME_US, ul_tmp) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_long(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_DATA_BYTES, p_st_l4_trans->l_request_bytes) == FR_FAIL)
            goto ERROR;

        p_st_l4_trans->l_request_bytes = (long)p_st_session_private->p_st_data->ui_remaining;
        if(p_st_private->is_message)
            p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_REQ);
        else
            p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_SUB);

        if(p_st_report_data == NULL)
            goto ERROR;

        p_st_private->p_lst_comm_tmp = FRREPORT_BODY_L7LIST_CLONE_COMM(p_st_private->p_st_body_data_req);
        if(p_st_private->p_lst_comm_tmp == NULL)
            goto ERROR;

        p_st_private->p_lst_req_tmp = FRREPORT_BODY_L7LIST_CLONE_REQ(p_st_private->p_st_body_data_req);
        if(p_st_private->p_lst_req_tmp == NULL)
            goto ERROR;

        p_st_private->p_lst_mq_tmp = FRREPORT_BODY_L7LIST_CLONE_MQ(p_st_private->p_st_body_data_req);
        if(p_st_private->p_lst_mq_tmp == NULL)
            goto ERROR;

        if(frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_req) == FR_FAIL)
            goto ERROR;
        p_st_private->p_st_body_data_req = NULL;

        if(frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
            goto ERROR;

        if(p_st_session_private->i_version == 10)
        {
            p_st_header = &p_st_session_private->st_session_private;

            p_st_node = frlist_node_create(p_st_private);
            if(p_st_node == NULL)
                return FR_FAIL;

            frlist_push(p_st_header, p_st_node);
            p_st_session_private->p_st_private = NULL;
        }

        if(p_st_session_private->i_version == 9)
        {
            __amqp_session_private_destroy (p_st_session_private->p_st_private);
            p_st_session_private->p_st_private = NULL;
        }

    }

    if(p_st_session_private->p_st_data->ui_remaining)
    {
        i_falg_analysis = 0;
        goto ANALYSIS;
    }

    //数据删除
    bytebuff_free(&p_st_session_private->p_st_data);
    p_st_session_private->p_st_data = bytebuff_creat();
    if(p_st_session_private->p_st_data == NULL)
        return APM_L7_ANALYZE_FAIL;

    if(i_ret == APM_L7_ANALYZE_SUCC)
        return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);

    return i_ret;

ERROR:
    if(p_st_report_data)
        frreport_data_free(&p_st_report_data);

    if(p_st_private->p_lst_comm_tmp)
        frreport_body_l7list_free(&p_st_private->p_lst_comm_tmp);

    if(p_st_private->p_lst_req_tmp)
        frreport_body_l7list_free(&p_st_private->p_lst_req_tmp);

    if(p_st_private->p_lst_mq_tmp)
        frreport_body_l7list_free(&p_st_private->p_lst_mq_tmp);

    return APM_L7_ANALYZE_FAIL;
}


/*******************************************************************************
*  FUNC     :  amqp resp 的packet解析
*  ARGS     :  p_st_conn                (入力链接信息)
            :  p_st_packet             (入力packet信息)
            :  p_st_hash_report         (保存Frreport_data)
*  RTN      :  APM_L7_ANALYZE_SUCC         成功
*           :  APM_L7_ANALYZE_TODO         数据未结束
*           :  APM_L7_ANALYZE_FAIL         失败
*  NOTE     :
*******************************************************************************/
FR_STATIC int __resp_data_analysis(Conntrack *p_st_conn
        , Apm_packet *p_st_packet
        , Frhash_header *p_st_hash_report
        , L7_session *p_st_session)
{
    int i_ret = 0;
    st_amqp_private_data *p_st_private = NULL;
    L7session_amqp_private *p_st_session_private = NULL;
    unsigned long ul_tmp = 0;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data *p_st_report_data = NULL;
//    Frlist_node *p_st_node = NULL;
    Frlist_header *p_st_header = NULL;
    int i_id_first = 0, i_id_last = 0;
    char arr_info[1024];
    int i_analysis = 1, i_temp_pos = 0;

    if(p_st_conn == NULL || p_st_packet == NULL
       || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

ANALYSIS:
    p_st_session_private = (L7session_amqp_private*)p_st_session->p_session_private;
    if(p_st_session_private == NULL)
    {
        p_st_session->p_session_private = __amqp_session_private_create();
        p_st_session_private = (L7session_amqp_private*) p_st_session->p_session_private;
    }

    //添加到bytebyff
    if(i_analysis)
    {
        i_ret = bytebuff_add(p_st_session_private->p_st_data, p_st_packet->p_payload, p_st_packet->ui_payload_len);
        if(i_ret == FR_FAIL)
            return APM_L7_ANALYZE_FAIL;
    }

    if(p_st_session_private->p_st_private == NULL && p_st_session_private->i_version == 10)
    {
        i_temp_pos = bytebuff_getPosition(p_st_session_private->p_st_data);

        p_st_header = &p_st_session_private->st_session_private;
        //获取id
        i_ret = amqp_get_disposition_id(p_st_session_private->p_st_data, &i_id_first, &i_id_last);
        if(i_ret == FR_SUCC)
        {
            //p_st_private 获取
            p_st_private = __session_data_get(p_st_header, &i_id_first, &i_id_last);
            if(p_st_private == NULL)
            {
                p_st_session_private->p_st_private = __amqp_private_create ();
                if(p_st_session_private->p_st_private == NULL)
                    return APM_L7_ANALYZE_FAIL;

                p_st_private = p_st_session_private->p_st_private;
            }
        }
        else
        {
            p_st_session_private->p_st_private = __amqp_private_create ();
            if(p_st_session_private->p_st_private == NULL)
                return APM_L7_ANALYZE_FAIL;

            p_st_private = p_st_session_private->p_st_private;
        }

        bytebuff_setPosition(p_st_session_private->p_st_data, i_temp_pos);
    }
    else
    {
        if(p_st_session_private->p_st_private == NULL)
            p_st_session_private->p_st_private = __amqp_private_create ();

        if(p_st_session_private->p_st_private == NULL)
            return APM_L7_ANALYZE_FAIL;

        p_st_private = p_st_session_private->p_st_private;
    }


    p_st_l4_trans = &(p_st_private->st_l4_trans);
    p_st_private->i_use_request = FR_NO;

    p_st_l4_trans->l_response_bytes = (long)bytebuff_remaining(p_st_session_private->p_st_data);

    if (p_st_l4_trans->st_response_time_first_byte.tv_sec == 0 &&
        p_st_l4_trans->st_response_time_first_byte.tv_usec == 0)
    {
        p_st_private->p_st_body_data_resp = frreport_body_new_l7(p_st_conn);
        if(p_st_private->p_st_body_data_resp == NULL)
            return APM_L7_ANALYZE_FAIL;

        //应答发起时间戳
        p_st_l4_trans->st_response_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
        p_st_l4_trans->st_response_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;

        //首包时间
        if(p_st_l4_trans->st_request_time_first_byte.tv_sec != 0)
            p_st_l4_trans->ul_response_time_firstbyte_us = (unsigned long)frtime_diff(&(p_st_l4_trans->st_request_time_first_byte), &(p_st_l4_trans->st_response_time_first_byte));
        else
            p_st_l4_trans->ul_response_time_firstbyte_us = 0;

    }
    //st_response_time
    p_st_l4_trans->st_response_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
    p_st_l4_trans->st_response_time.tv_usec = p_st_packet->st_timestamp.tv_usec;
    /* 请求的字节数 */

    //如果skip长度有剩余  返回TODO
    if((p_st_session_private->p_st_data)->ui_skip_len)
        return APM_L7_ANALYZE_TODO;

    i_ret = amqp_parse(p_st_session_private->p_st_data, p_st_private->p_st_body_data_resp,p_st_private, p_st_session_private);
    if(i_ret == APM_L7_ANALYZE_FAIL)
        return APM_L7_ANALYZE_FAIL;

    if(p_st_session_private->i_flag_data_type)
    {
        if (p_st_session_private->i_flag_data_type == FRLOGIN_SUCC)
        {
            if (p_st_session_private->p_user_name != NULL && *p_st_session_private->p_user_name != '\0')
                snprintf(arr_info, sizeof(arr_info), "%s login", p_st_session_private->p_user_name);
            else
                snprintf(arr_info, sizeof(arr_info), "login");

            p_st_report_data = frreport_update_create(p_st_conn, APM_REPORT_UPDATE_LOGON, arr_info, &p_st_packet->st_timestamp);

        }
        else if (p_st_session_private->i_flag_data_type == FRLOGIN_FAIL)
        {
            if (p_st_session_private->p_user_name != NULL && *p_st_session_private->p_user_name != '\0')
                snprintf(arr_info, sizeof(arr_info), "%s login reject", p_st_session_private->p_user_name);
            else
                snprintf(arr_info, sizeof(arr_info), "login reject");

            p_st_report_data = frreport_update_create(p_st_conn, APM_REPORT_UPDATE_LOGON_REJECT, arr_info, &p_st_packet->st_timestamp);

        }
        else if(p_st_session_private->i_flag_data_type == FRLOGOUT_END)
        {

            if (p_st_session_private->p_user_name != NULL && *p_st_session_private->p_user_name != '\0')
                snprintf(arr_info, sizeof(arr_info), "%s logout", p_st_session_private->p_user_name);
            else
                snprintf(arr_info, sizeof(arr_info), "logout");

            p_st_report_data = frreport_update_create(p_st_conn, APM_REPORT_UPDATE_LOGOUT, arr_info, &p_st_packet->st_timestamp);

        }

        p_st_session_private->i_flag_data_type = 0;
        if (p_st_report_data)
        {
            if(frreport_send(g_p_apm_context, &p_st_report_data) == FR_FAIL)
            {
                frreport_data_free(&p_st_report_data);
                return APM_L7_ANALYZE_FAIL;
            }
        }
    }

    //数据上报
    if(i_ret == APM_L7_ANALYZE_SUCC && p_st_private->i_flag_report)
    {
        p_st_l4_trans->l_response_bytes -= (long)p_st_session_private->p_st_data->ui_remaining;

        if(p_st_private->ul_seq == 0 || p_st_private->p_lst_comm_tmp == NULL)
            p_st_private->ul_seq = frct_seq_increase(p_st_conn);

        //只有response
        if(p_st_private->p_lst_comm_tmp == NULL)
        {
            if(frreport_body_l7_add_connInfo(p_st_private->p_st_body_data_resp, p_st_conn) == FR_FAIL)
                goto ERROR;

            if(frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_SERVER_TIME_US, 0) == FR_FAIL)
                goto ERROR;

            if(__amqp_session_report(p_st_private->p_st_body_data_resp, p_st_session_private) == FR_FAIL)
                goto ERROR;

            if(__amqp_title_report(p_st_private->p_st_body_data_resp, p_st_private) == FR_FAIL)
                goto ERROR;

        }
        else
        {
            //comm
            if(FRREPORT_BODY_L7LIST_SET_COMM(p_st_private->p_st_body_data_resp, &p_st_private->p_lst_comm_tmp) == FR_FAIL)
                goto ERROR;

            //把req数据链表添加到resp中
            if(FRREPORT_BODY_L7LIST_SET_REQ(p_st_private->p_st_body_data_resp, &p_st_private->p_lst_req_tmp) == FR_FAIL)
                goto ERROR;

            if(p_st_private->is_topic)
            {
                if(FRREPORT_BODY_L7LIST_SET_MQ(p_st_private->p_st_body_data_resp, &p_st_private->p_lst_mq_tmp) == FR_FAIL)
                    goto ERROR;
            }

            ul_tmp = (unsigned long)frtime_diff(&p_st_l4_trans->st_request_time, &p_st_l4_trans->st_response_time_first_byte);
            if(frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_SERVER_TIME_US, ul_tmp) == FR_FAIL)
                goto ERROR;

        }

        if(frreport_body_l7_add_int(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_START_TIMESTAMP, &p_st_l4_trans->st_response_time_first_byte) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_END_TIMESTAMP, &p_st_l4_trans->st_response_time) == FR_FAIL)
            goto ERROR;

        ul_tmp = (unsigned long)frtime_diff(&p_st_l4_trans->st_response_time_first_byte, &p_st_l4_trans->st_response_time);
        if(frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DOWNLOAD_TIME_US, ul_tmp) == FR_FAIL)
            goto ERROR;

        if(frreport_body_l7_add_long(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DATA_BYTES, p_st_l4_trans->l_response_bytes) == FR_FAIL)
            goto ERROR;

        p_st_l4_trans->l_response_bytes = (long)p_st_session_private->p_st_data->ui_remaining;
        if(p_st_private->is_message)
            p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_NOTIFY);
        else
            p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_RESP);

        if(p_st_report_data == NULL)
            goto ERROR;

        if(frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_resp) == FR_FAIL)
            goto ERROR;
        p_st_private->p_st_body_data_resp = NULL;

        if(frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
            goto ERROR;

    }

    if(p_st_private->i_refence == 0 || p_st_session_private->i_version == 9)
    {
        if(__amqp_session_private_destroy(p_st_private) == FR_FAIL)
            return APM_L7_ANALYZE_FAIL;

        if(p_st_session_private->p_st_private)
            p_st_session_private->p_st_private = NULL;
    }
    else
    {
        p_st_l4_trans->st_response_time_first_byte.tv_sec = 0;
        p_st_l4_trans->st_response_time_first_byte.tv_usec = 0;
        p_st_l4_trans->l_response_bytes  = 0;
    }

    if(p_st_session_private->p_st_data->ui_remaining)
    {
        i_analysis = 0;
        goto ANALYSIS;
    }

    //数据删除
    bytebuff_free(&p_st_session_private->p_st_data);
    p_st_session_private->p_st_data = bytebuff_creat();
    if(p_st_session_private->p_st_data == NULL)
        return APM_L7_ANALYZE_FAIL;

    if(i_ret == APM_L7_ANALYZE_SUCC)
        return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);

    return i_ret;

ERROR:
    if(p_st_report_data)
        frreport_data_free(&p_st_report_data);

    return APM_L7_ANALYZE_FAIL;

}


/*******************************************************************************
*  FUNC     :  amqp的packet解析
*  ARGS     :  p_st_conn                (入力链接信息)
            :  p_st_packet             (入力packet信息)
            :  p_st_hash_report         (保存Frreport_data)
*  RTN      :  APM_L7_ANALYZE_SUCC         成功
*           :  APM_L7_ANALYZE_TODO         数据未结束
*           :  APM_L7_ANALYZE_FAIL         失败
*  NOTE     :
*******************************************************************************/
int amqp_analysis(Conntrack *p_st_conn
        , Apm_packet *p_st_packet
        , Frhash_header *p_st_hash_report)
{
    int i_ret = 0;
    L7_session *p_st_session = NULL;

    if(p_st_conn == NULL || p_st_packet == NULL
       || p_st_hash_report == NULL)
        return APM_L7_ANALYZE_FAIL;

    /* 到Conntrack中取得session信息 */
    p_st_session = frct_get_session(p_st_conn);
    if (p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    //fin syn 包  销毁私有数据
    if(p_st_packet->em_analyz_cmd == FRANALYZ_CMD_FIN || p_st_packet->em_analyz_cmd == FRANALYZ_CMD_SYN)
    {
        amqp_session_destroy(p_st_session->p_session_private);
        p_st_session->p_session_private = NULL;
        return APM_L7_ANALYZE_SUCC;
    }

    //p_session_private   分配内存
    if(p_st_session->p_session_private == NULL)
    {
        p_st_session->p_session_private = __amqp_session_private_create();
        //内存分配失败
        if(p_st_session->p_session_private == NULL)
            return APM_L7_ANALYZE_FAIL;
    }

    //request
    if(p_st_packet->i_direction == FR_DIRECT_CLIENT2SRV)
    {
        i_ret = __req_data_analysis(p_st_conn,p_st_packet,p_st_hash_report,p_st_session);
        if(i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;

        return i_ret;
    }
    else//response
    {
        i_ret = __resp_data_analysis(p_st_conn,p_st_packet,p_st_hash_report, p_st_session);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;

        return i_ret;
    }

ERROR:

    amqp_session_destroy(p_st_session->p_session_private);
    p_st_session->p_session_private = NULL;
    return APM_L7_ANALYZE_FAIL;

}


FR_STATIC void __amqp_list_node_free(Frlist_node *p_node)
{
    st_amqp_private_data *p_st_private = NULL;

    if(p_node && p_node->data)
    {
        p_st_private = (st_amqp_private_data* )p_node->data;
        __amqp_session_private_destroy(p_st_private);
    }

    return;
}


/*******************************************************************************
*  FUNC     :  私有信息销毁
*  ARGS     :  p_private_session(amqp的私有session数据)
*  RTN      :  FR_SUCC
*  NOTE     :
*******************************************************************************/
int amqp_session_destroy(void *p_private_session)
{
    L7session_amqp_private *p_st_session_private = (L7session_amqp_private* )p_private_session;

    if(p_st_session_private)
    {
        if(p_st_session_private->p_user_name)
            frfree(p_st_session_private->p_user_name);

        if(p_st_session_private->p_server)
            frfree(p_st_session_private->p_server);

        if(p_st_session_private->p_st_data)
            bytebuff_free(&p_st_session_private->p_st_data);

        if(p_st_session_private->p_st_private)
            __amqp_session_private_destroy(p_st_session_private->p_st_private);

        frlist_remove_all(&p_st_session_private->st_session_private, __amqp_list_node_free);

        frfree(p_st_session_private);
    }

    return FR_SUCC;
}


